package cn.xinfei.xdecision.engine.runner.consumer;

import cn.xinfei.xdecision.common.utils.util.IpUtil;
import cn.xinfei.xdecision.common.utils.util.JsonUtils;
import cn.xinfei.xdecision.constant.GlobalConstant;
import cn.xinfei.xdecision.engine.runner.executor.model.EngineNodeLog;
import cn.xinfei.xdecision.engine.runner.service.DecisionResultDetailService;
import cn.xinfei.xdecision.pulsar.config.XDecisionMessage;
import cn.xinfei.xdecision.pulsar.consumer.XPulsarConsumerAnnotate;
import com.alibaba.fastjson.JSONObject;
import com.dianping.cat.Cat;
import com.dianping.cat.message.Transaction;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Tags;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.PulsarClientException;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;

@Service
@Slf4j
public class XDecisionNodeLogConsumer {


    @Autowired
    private DecisionResultDetailService decisionResultDetailService;

    @Autowired
    private MeterRegistry meterRegistry;

    /**
     * 推荐手动提交ack
     *
     * @param consumer
     * @return
     */
//    @XPulsarConsumerAnnotate(
//            topic = "${xdecision.decisionresult.nodeLogTopic}",
//            clazz = String.class,
//            consumerName = "${spring.xpulsar.consumer.consumerName}",
//            subscriptionName = "${spring.xpulsar.consumer.subscriptionName}",
//            concurrent = "${spring.xpulsar.consumer.concurrent}",
//            batch = false)
    public List<MessageId> consume(Message<String> message, Consumer<String> consumer) {

        Transaction transaction = Cat.newTransaction("xengine", "nodeLogTopic");
//            log.info("Received {} messages in a single batch receive.", messages.size());
//            for (Message<String> message : messages) {
        try {
            String msg = message.getValue();
            String key = message.getKey();
            XDecisionMessage xDecisionMessage = JSONObject.parseObject(msg, XDecisionMessage.class);
            String msgId = xDecisionMessage.getMsgId();
            MessageId messageId = message.getMessageId();
            MDC.put(GlobalConstant.TRACE_ID, msgId);
            MDC.put(GlobalConstant.PID, key);
            String data = xDecisionMessage.getData();
            EngineNodeLog engineNodeLog = JSONObject.parseObject(data, EngineNodeLog.class);
            engineNodeLog.getIps().push("c_" + IpUtil.getIpAddress());

            meterRegistry.counter("xdecision_consumer_total", Tags.of("topic", message.getTopicName())).increment();
            log.info("nodeLog key={},msgId={},messageId={}", key, msgId, messageId);
            boolean resultDetail = decisionResultDetailService.saveEngineNodeLog(JsonUtils.toJsonString(engineNodeLog));
            if (resultDetail) {
                consumer.acknowledgeAsync(messageId)
                        .handle((result, ex) -> {
                            if (ex == null) {
                                log.info("nodeLog ack success, key={},msgId={},messageId={},resultDetail={}", key, msgId, messageId.toString(), resultDetail);
//                                        successHandler.accept(messageId);
                                meterRegistry.counter("xdecision_consumer_total", Tags.of("topic", message.getTopicName(),"ack","success")).increment();
                            } else {
                                log.error("nodeLog ack fail, key={},msgId={},messageId={},resultDetail={}", key, msgId, messageId.toString(), resultDetail, ex);
//                                        failHandler.accept((Exception) ex);
                                meterRegistry.counter("xdecision_consumer_total", Tags.of("topic", message.getTopicName(),"ack","fail")).increment();
                            }
                            MDC.clear();
                            return true;
                        });
            } else {
                log.error("nodeLog fail. key={},msgId={},msgId={},resultDetail={}", key, msgId, messageId.toString(), resultDetail);
                consumer.negativeAcknowledge(messageId);
            }
            transaction.setStatus(com.dianping.cat.message.Message.SUCCESS);
        } catch (ClassCastException e) {
            log.error("logConsumerFail,类型转换异常,提交ack,msg={}", e.getMessage(), e);
            try {
                consumer.acknowledge(message);
            } catch (PulsarClientException pulsarClientException) {
                pulsarClientException.printStackTrace();
            }
            Cat.logError(e);
            meterRegistry.counter("xdecision_consumer_total", Tags.of("topic", message.getTopicName(),"exception","ClassCastException")).increment();


        } catch (Exception e) {
            log.error("logConsumerFail,msg={}", e.getMessage(), e);
            consumer.negativeAcknowledge(message);
            Cat.logError(e);
            meterRegistry.counter("xdecision_consumer_total", Tags.of("topic", message.getTopicName(),"exception","Exception")).increment();

        } finally {

            MDC.clear();
        }
//            }
        transaction.complete();
//        }

        return null;
    }




}
